package com.accenture.automation.biz.api.producer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import com.accenture.automation.biz.api.bean.ResponseBaseBean;
import com.accenture.automation.biz.api.enums.EnumWebServiceOutputStatus;
import com.accenture.automation.biz.api.producer.iface.ProducerService;
import com.accenture.automation.common.ApiClientUtil;

@Service
public class ProducerServiceImpl implements ProducerService {
	protected final Logger logger = LoggerFactory.getLogger(this.getClass());

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	@Override
	public int sendKafka(String topic, String key, String message) {
		try {
			// send如果传入key,kafka自动对 key 进行hash 选出一个 patition
			// send如果没传key,kafka自动轮询选出一个 patition
			kafkaTemplate.send(topic, key, message);
			logger.info("发送kafka成功,---------TOPIC:" + topic + "---------KEY:" + key + "----------MESSAGE:" + message);
			return EnumWebServiceOutputStatus.WEB_SERVICE_OUTPUT_SUCCESS.getApiCode();
		} catch (Exception e) {
			logger.error("发送kafka失败", e.getMessage());
			return EnumWebServiceOutputStatus.WEB_SERVICE_OUTPUT_SYSTEM_ERROR.getApiCode();
		}
	}

	@Override
	public ResponseBaseBean<Object> apiCreate(String topic, String message, String key) {
		ResponseBaseBean<Object> rsBean = new ResponseBaseBean<>();
		// 请求内容为空，返回错误
		if (StringUtils.isEmpty(message)) {
			ApiClientUtil.setErrorResponse(rsBean, String.valueOf(9999), "request body is null");
			return rsBean;
		}
		try {
			int apiCode = sendKafka(topic, key, message);
			ApiClientUtil.setAccountPostResult(rsBean, apiCode);
		} catch (Exception e) {
			ApiClientUtil.setErrorResponse(rsBean, String.valueOf(9999), e.getMessage());
		}
		return rsBean;
	}

}